package consumers

import (
	"app-api/clients/video"
	"app-api/common/constant"
	"context"
	"strconv"

	"strings"

	"app-api/app/internal/svc"
	"github.com/apache/pulsar-client-go/pulsar"
	"github.com/zeromicro/go-zero/core/logx"
)

type ConsumerLogic struct {
	logx.Logger
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

func NewDConsumerLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ConsumerLogic {
	return &ConsumerLogic{
		Logger: logx.WithContext(ctx),
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func UserVideoConsumer(l *ConsumerLogic, topicKey string) {
	consumer, err := l.svcCtx.PulsarClient.Subscribe(pulsar.ConsumerOptions{
		Topic:            topicKey,
		SubscriptionName: "my-subscription",
		Type:             pulsar.Shared,
	})
	if err != nil {
		l.Logger.Error("Failed to create consumer: %v", err)
		return
	}

	msg, err := consumer.Receive(context.Background())
	if err != nil {
		l.Logger.Errorf("Failed to receive message: %v", err)
		return
	}

	topicVal := string(msg.Payload())
	l.Logger.Infof("Received message: %s with ID: %s", topicKey, msg.ID())
	err = consumer.Ack(msg)
	if err != nil {
		l.Logger.Errorf("Failed to ack message: %v", err)
		return
	}

	stringList := strings.Split(topicVal, ":")
	if len(stringList) != 2 {
		return
	}

	userId, _ := strconv.ParseInt(stringList[0], 10, 64)
	videoId, _ := strconv.ParseInt(stringList[1], 10, 64)
	switch {
	case strings.HasPrefix(topicKey, constant.UserLikeVideoTopic):
		_, err = l.svcCtx.VideoRPC.LikeVideo(l.ctx, &video.FavVideoRequest{UserId: userId, VideoId: videoId})
		if err != nil {
			l.Logger.Errorf("Failed to unlike video: %v", err)
		}
	case strings.HasPrefix(topicKey, constant.UserUnLikeVideoTopic):
		_, err = l.svcCtx.VideoRPC.UnlikeVideo(l.ctx, &video.FavVideoRequest{UserId: userId, VideoId: videoId})
		if err != nil {
			l.Logger.Errorf("Failed to unlike video: %v", err)
		}
	}
}
